package com.ds.infrastructure.hyperspace.container.endpoint.connector.health;

import com.ds.infrastructure.hyperspace.common.config.HealthCheckConfig;
import com.ds.infrastructure.hyperspace.common.config.TargetProtocolConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ScheduledFuture;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicBoolean;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * @author  Weishaoying
 */
@Slf4j
public class HyperspaceNodeHealthChecker implements Runnable {

    private final NettyContainer nettyContainer;

    private final TargetEndpointNode targetEndpointNode;

    private volatile ScheduledFuture<?> scheduledFuture;

    private AtomicBoolean isChecking = new AtomicBoolean();

    public HyperspaceNodeHealthChecker(NettyContainer nettyContainer,
                                       TargetEndpointNode targetEndpointNode) {
        this.nettyContainer = nettyContainer;
        this.targetEndpointNode = targetEndpointNode;
        HealthCheckConfig healthCheckConfig = targetEndpointNode.getNodeConfig()
                .getProtocolConfig().getHealthCheckConfig();
        this.scheduledFuture = nettyContainer.getTargetHealthCheckEventLoopGroup().scheduleWithFixedDelay(
                this, healthCheckConfig.getInterval(), healthCheckConfig.getInterval(), MILLISECONDS);
    }

    public void reschedule() {
        try {
            if (scheduledFuture != null) {
                scheduledFuture.cancel(true);
            }
        } catch (Exception e) {
            log.error("Cancel health check failed!", e);
        }
        try {
            HealthCheckConfig healthCheckConfig = targetEndpointNode.getNodeConfig()
                    .getProtocolConfig().getHealthCheckConfig();
            scheduledFuture = nettyContainer.getTargetHealthCheckEventLoopGroup().scheduleWithFixedDelay(
                    this, healthCheckConfig.getInterval(), healthCheckConfig.getInterval(), MILLISECONDS);
        } catch (Exception e) {
            log.error("Cancel health check failed!", e);
        }
    }

    public void cancel() {
        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }
    }

    @Override
    public void run() {
        if (isChecking.get() || !isChecking.compareAndSet(false, true)) {
            return;
        }
        boolean isAutoComplete = true;
        try {
            TargetProtocolConfig protocolConfig = targetEndpointNode.getNodeConfig().getProtocolConfig();
            for (InetSocketAddress remoteAddress : targetEndpointNode.getResolverProvider().getAllNode()) {
                Bootstrap bootstrap = new Bootstrap().group(nettyContainer.getTargetHealthCheckEventLoopGroup())
                        .channel(nettyContainer.getNettyTransportFactoryDelegate().socketChannelClass())
                        .option(ChannelOption.TCP_NODELAY, true)
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, protocolConfig.getConnectTimeout())
                        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                        .handler(new HealthCheckChannelInitializer(targetEndpointNode, this))
                        .remoteAddress(remoteAddress);
                bootstrap.connect().addListener(new HealthCheckConnectFutureListener(
                        targetEndpointNode, this, remoteAddress));
                isAutoComplete = false;
            }
        } catch (Exception ex) {
            log.error("Run Http health check failed! ", ex);
        } finally {
            if (isAutoComplete) {
                checkComplete();
            }
        }

    }

    public void checkComplete() {
        isChecking.compareAndSet(true, false);
    }

    static class HealthCheckConnectFutureListener implements GenericFutureListener<ChannelFuture> {

        private final TargetEndpointNode targetEndpointNode;

        private final HyperspaceNodeHealthChecker nodeHealthChecker;

        private final InetSocketAddress remoteAddress;

        public HealthCheckConnectFutureListener(TargetEndpointNode targetEndpointNode,
                                                HyperspaceNodeHealthChecker nodeHealthChecker,
                                                InetSocketAddress remoteAddress) {
            this.targetEndpointNode = targetEndpointNode;
            this.nodeHealthChecker = nodeHealthChecker;
            this.remoteAddress = remoteAddress;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                targetEndpointNode.incrementFailCount(remoteAddress);
                nodeHealthChecker.checkComplete();
                future.channel().close();
                String errorMsg = String.format("Connect to target failed,target info(%s),remote address:%s",
                        targetEndpointNode.getKey(), remoteAddress);
                log.error(errorMsg, future.cause());
            }
        }
    }
}
